package com.lenovo.spider;

import cn.hutool.core.thread.ThreadUtil;
import com.lenovo.spider.component.TaskIterator;
import com.lenovo.spider.component.TaskProcess;
import com.lenovo.spider.component.TaskProcessor;
import com.lenovo.spider.domain.Task;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import reactor.core.publisher.WorkQueueProcessor;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

import static com.lenovo.spider.common.Functions.monoWrap;
import static com.lenovo.spider.common.Functions.when;

/**
 * @author <a href="mailto:liuhan3@lenovo.com">HanL(liuhan3)</a>
 * on 2018/4/11 18:46
 */
@Slf4j
public class Spider {
    public static final int WAIT = 0;
    public static final int RUNNING = 1;
    public static final int STOP = 2;
    /**
     * 工作流处理器
     */
    private final WorkQueueProcessor<Task> processor;
    /**
     * 任务迭代器
     */
    private TaskIterator taskIterator;

    /**
     * 下载器
     */
    private TaskProcessor<TaskProcess.Downloader> downloader;
    /**
     * 页面解析器
     */
    private TaskProcessor<TaskProcess.PageParser> pageParser;
    /**
     * 结果处理管道
     */
    private TaskProcessor<TaskProcess.Pipeline> pipeline;
    /**
     * 异常处理器
     */
    private TaskProcessor<TaskProcess.ExceptionHandler> exceptionHandler;

    private TaskProcessor<TaskProcess.PreDownloaderInterceptor> preInterceptor;

    private TaskProcessor<TaskProcess.AfterDownloaderInterceptor> afterInterceptor;

    /**
     * 爬虫状态
     */
    private AtomicInteger state;

    public Spider(TaskIterator taskIterator) {
        this.taskIterator = taskIterator;
        val spiderName = ConfigHolder.getString("spider.name", Spider.class.getSimpleName());
        val bufferSize = ConfigHolder.getInt("spider.bufferSize", 1024);
        this.processor = WorkQueueProcessor.create(spiderName, bufferSize);
        this.state = new AtomicInteger(0);
        this.downloader = TaskProcessor.downloader();
        this.pageParser = TaskProcessor.pageParser();
        this.pipeline = TaskProcessor.pipeline();
        this.exceptionHandler = TaskProcessor.exceptionHandler();
        this.preInterceptor = TaskProcessor.preInterceptor();
        this.afterInterceptor = TaskProcessor.afterInterceptor();
        configProcessor();
    }

    /**
     * 配置爬虫处理工作流
     */
    private void configProcessor() {
        this.processor.delayElements(Duration.ofMillis(500))
                .log()
                .flatMap(monoWrap(this.preInterceptor::process))
                .flatMap(monoWrap(this.downloader::process))  // 下载
                .flatMap(monoWrap(this.afterInterceptor::process))
                .flatMap(monoWrap(this.pageParser::process)) // 页面解析
                .subscribe(when(Task::isSuccess,
                        this.pipeline::process, // 结果处理
                        this.exceptionHandler::process  //异常处理
                ), throwable -> log.error(throwable.getMessage(), throwable));
    }

    /**
     * 获取爬虫执行体
     *
     * @return
     */
    public Runnable getRunnable() {
        return this::run;
    }

    public void run() {
        this.setRunning(); // 设置启动状态
        log.info("爬虫线程启动了");
        while (this.isRunning() && this.taskIterator.hasNext()) {
            try {
                ThreadUtil.sleep(500);
                this.processor.onNext(this.taskIterator.next());
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
        // 爬虫停止后，等待流处理器中的处理完成后再关闭流处理器
        this.processor.awaitAndShutdown();
//        this.processor.forceShutdown()
//                .subscribe(this.taskIterator::addTask)
//        ;
    }

    /**
     * 停止
     */
    public void stop() {
        for (; ; ) {
            if (this.state.compareAndSet(RUNNING, STOP)) {

                break;
            }
        }
    }

    /**
     * 判断当前状态是否正在运行
     *
     * @return
     */
    public boolean isRunning() {
        return this.state.get() == RUNNING;
    }

    /**
     * 判断当前状态是否未运行
     *
     * @return
     */
    public boolean isNotRunning() {
        return this.state.get() != RUNNING;
    }

    /**
     * 设置运行状态
     */
    private void setRunning() {
        for (; ; ) {
            if (this.state.compareAndSet(WAIT, RUNNING)) {
                break;
            }
        }
    }

}
